package com.yuyunbo.hbase.autoconfigure;

import com.yuyunbo.hbase.HbaseApiService;
import com.yuyunbo.hbase.HbaseModel;
import com.yuyunbo.hbase.constants.Constants;
import com.yuyunbo.hbase.util.HbaseAnnotationUtil;
import com.yuyunbo.hbase.util.HbaseBeanUtils;
import com.yuyunbo.hbase.util.HbaseException;
import com.yuyunbo.hbase.util.MapperUtils;
import com.yuyunbo.hbase.vo.HbasePageBean;
import com.yuyunbo.hbase.vo.HbaseQueryVo;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @Author: yuyunbo-1154363414@qq.com
 * @Date: 2020/5/8 15:21
 * @Description:
 */
public class HbaseApiServiceImpl implements HbaseApiService {
    @Autowired
    private HbaseClient hbaseClient;
    private static final String ERRORSTR = "推送数据为空";
    private static final String ANNOERRORSTR = "缺少注解(HbTableName/HbRowKey)";
    private static final String QUERYERROR = "查询数据缺少必传字段";
    private static final String COLUMNNAMEERROR = "查询数据列集合为空";
    private static final String ROWKEYEERROR = "查询数据主键集合为空";

    /**
     * 单个提交实体数据到Hbase
     *
     * @param hbaseModel
     */
    @Override
    public <T> void singlePut(HbaseModel<T> hbaseModel) {
        if (hbaseModel.getData() == null) {
            throw new HbaseException(ERRORSTR);
        }
        String hbcorkString = HbaseAnnotationUtil.getHbaseTbCoRkValue(hbaseModel.getData());
        String[] hbcorkStringArray = hbcorkString.split(Constants.COMMA);
        if (hbcorkStringArray.length != 3) {
            throw new HbaseException(ANNOERRORSTR);
        }
        String tbName = hbcorkStringArray[0];
        String columnFamily = hbcorkStringArray[1];
        String rowKey = hbcorkStringArray[2];
        hbaseClient.singlePut(tbName, rowKey, columnFamily, HbaseBeanUtils.beanToMap(hbaseModel.getData()));
    }

    /**
     * 批量提交数据 到habse
     *
     * @param hbaseModel
     */
    @Override
    public <T> void multiplePut(HbaseModel<T> hbaseModel) {
        if (CollectionUtils.isEmpty(hbaseModel.getPutList())) {
            throw new HbaseException(ERRORSTR);
        }
        String hbcorkString = HbaseAnnotationUtil.getHbaseTbCoRkValue(hbaseModel.getPutList().get(0));
        String[] hbcorkStringArray = hbcorkString.split(Constants.COMMA);
        if (hbcorkStringArray.length != 3) {
            throw new HbaseException(ANNOERRORSTR);
        }
        String tbName = hbcorkStringArray[0];
        String columnFamily = hbcorkStringArray[1];
        List<Map<String, String>> putListMap = new ArrayList<>();
        hbaseModel.getPutList().forEach(data -> {
            putListMap.add(HbaseBeanUtils.beanToMap(data));
        });
        hbaseClient.multiplePut(tbName, columnFamily, putListMap);
    }

    /**
     * 批量异步put操作
     *
     * @param hbaseModel
     */
    @Override
    public <T> void multiplePutAsync(HbaseModel<T> hbaseModel) {
        if (CollectionUtils.isEmpty(hbaseModel.getPutList())) {
            throw new HbaseException(ERRORSTR);
        }
        String hbcorkString = HbaseAnnotationUtil.getHbaseTbCoRkValue(hbaseModel.getPutList().get(0));
        String[] hbcorkStringArray = hbcorkString.split(Constants.COMMA);
        if (hbcorkStringArray.length != 3) {
            throw new HbaseException(ANNOERRORSTR);
        }
        String tbName = hbcorkStringArray[0];
        String columnFamily = hbcorkStringArray[1];
        List<Map<String, String>> putListMap = new ArrayList<>();
        hbaseModel.getPutList().forEach(data -> {
            putListMap.add(HbaseBeanUtils.beanToMap(data));
        });
        hbaseClient.multiplePutAsync(tbName, columnFamily, putListMap);
    }



    /**
     * 根据查询条件,统计符合条件数据
     *
     * @param hbaseQueryVo
     * @param pageSize
     * @return
     */
    @Override
    public <T> HbasePageBean<T> queryListByParamForPage(HbaseQueryVo hbaseQueryVo, int pageSize, Class<T> requiredType) {
        String tbName = hbaseQueryVo.getTbName();
        String columnFamily = hbaseQueryVo.getColumnFamily();
        if (StringUtils.isEmpty(tbName) || StringUtils.isEmpty(columnFamily)) {
            throw new HbaseException(QUERYERROR);
        }
        HbasePageBean hbasePageBean = new HbasePageBean();
        List<Map<String, Object>> mapList = hbaseClient.queryListByParamForPage(tbName, columnFamily, hbaseQueryVo.getStartKey(), pageSize+1, hbaseQueryVo.getMatchParamMap());
        if (CollectionUtils.isNotEmpty(mapList)) {
            String jsonStr =  MapperUtils.obj2json(mapList);
            hbasePageBean.setDatas(MapperUtils.json2list(jsonStr,requiredType));
        }
        hbasePageBean.setPageSize(pageSize);
        return hbasePageBean;
    }

    /**
     * 表名,主键获取对应的数据
     *
     * @param hbaseQueryVo
     * @param requiredType
     * @return
     */
    @Override
    public <T> T singleGet(HbaseQueryVo hbaseQueryVo, Class<T> requiredType) {
        String tbName = hbaseQueryVo.getTbName();
        String rowKey = hbaseQueryVo.getRowKey();
        if (StringUtils.isEmpty(tbName) || StringUtils.isEmpty(rowKey)) {
            throw new HbaseException(QUERYERROR);
        }
        Map<String, Object> result = hbaseClient.singleGet(tbName, rowKey);
        if (result == null) {
            return null;
        }
        return MapperUtils.map2pojo(result,requiredType);
    }

    /**
     * 表名,主键获取对应列的数据
     *
     * @param hbaseQueryVo
     * @param requiredType
     * @return
     */
    @Override
    public <T> T singleGetWithColumn(HbaseQueryVo hbaseQueryVo, Class<T> requiredType) {
        String tbName = hbaseQueryVo.getTbName();
        String rowKey = hbaseQueryVo.getRowKey();
        String columnFamily = hbaseQueryVo.getColumnFamily();
        if (StringUtils.isEmpty(tbName) || StringUtils.isEmpty(rowKey) || StringUtils.isEmpty(columnFamily)) {
            throw new HbaseException(QUERYERROR);
        }
        if (CollectionUtils.isEmpty(hbaseQueryVo.getColumnNameList())) {
            throw new HbaseException(COLUMNNAMEERROR);
        }
        Map<String, Object> result = hbaseClient.singleGetWithColumn(tbName, rowKey, columnFamily, hbaseQueryVo.getColumnNameList());
        if (result == null) {
            return null;
        }
        return MapperUtils.map2pojo(result,requiredType);
    }

    /**
     * 根据表名,指定的主键集合,获取指定的列
     *
     * @param hbaseQueryVo
     * @param requiredType
     * @return
     */
    @Override
    public <T> List<T> multipleGet(HbaseQueryVo hbaseQueryVo, Class<T> requiredType) {
        String tbName = hbaseQueryVo.getTbName();
        String columnFamily = hbaseQueryVo.getColumnFamily();
        if (StringUtils.isEmpty(tbName) || StringUtils.isEmpty(columnFamily)) {
            throw new HbaseException(QUERYERROR);
        }
        if (CollectionUtils.isEmpty(hbaseQueryVo.getRowKeyList()) || CollectionUtils.isEmpty(hbaseQueryVo.getColumnNameList())) {
            throw new HbaseException(COLUMNNAMEERROR + Constants.SEPARATE_ + ROWKEYEERROR);
        }
        List<Map<String, Object>> result = hbaseClient.multipleGet(tbName, hbaseQueryVo.getRowKeyList(), columnFamily, hbaseQueryVo.getColumnNameList());
        if (CollectionUtils.isEmpty(result)) {
            return null;
        }
        String jsonStr =  MapperUtils.obj2json(result);
        return MapperUtils.json2list(jsonStr,requiredType);
    }

    /**
     * 获取主键匹配前缀的列,会存在全表扫描,性能存在风险
     *
     * @param hbaseQueryVo
     * @param requiredType
     * @return
     */
    @Override
    public <T> List<T> scanByPrefixFilter(HbaseQueryVo hbaseQueryVo, Class<T> requiredType) {
        String tbName = hbaseQueryVo.getTbName();
        String prefix = hbaseQueryVo.getPrefix();
        if (StringUtils.isEmpty(tbName) || StringUtils.isEmpty(prefix)) {
            throw new HbaseException(QUERYERROR);
        }
        List<Map<String, Object>> result = hbaseClient.scanByPrefixFilter(tbName, prefix);
        if (CollectionUtils.isEmpty(result)) {
            return null;
        }
        String jsonStr =  MapperUtils.obj2json(result);
        return MapperUtils.json2list(jsonStr,requiredType);
    }


}
